/*!
 * \file cassandra_core.c
 *
 * \brief 共通定义
 *
 * All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.\n
 * 共通使用的宏和结构体
 *
 * \version 1.0
 * \author jikw
 * \date 2016/1/22
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "config.h"
#include "frhash2.h"
#include "frct_comm.h"
#include "frreport_comm.h"
#include "frservice_comm.h"
#include "frbyte_buffer.h"
#include "cassandra_core.h"

extern void *g_p_apm_context;
struct timeval g_st_time;

/* 未上报的请求 */
typedef struct __st_cassandra_report_list {
    struct timeval st_request_time;
    Frlist_header *p_lst_comm;
    Frlist_header *p_lst_req;
} cassandra_report_list;

typedef struct __st_L7session_cassandra_private
{
    L4_transaction st_l4_trans;                       //数据
    Frreport_body_data *p_st_body_data_req;
    Frreport_body_data *p_st_body_data_resp;
    Frhash_header *p_report_list_cassandra;           //未上报的请求
    Frbyte_Packet *p_req_packet;                      //存请求bytebuff数据
    Frbyte_Packet *p_resp_packet;                     //存相应bytebuff数据
    uint16_t u16_req_stream;                          //req stream
    uint16_t u16_resp_stream;                         //resp stream
    int i_sign;                                       //event标志位
}L7session_cassandra_private;

/*!
 * cassandra私有数据创建
 *
 * \param void
 *
 * \return NULL          内存分配失败\n
 *         p_st_private  内存分配成功
 *
 * \note
 *
 */
FR_STATIC L7session_cassandra_private* __cassandra_private_create(void)
{
    L7session_cassandra_private *p_st_private = NULL;
    p_st_private = (L7session_cassandra_private*)frmalloc(sizeof(L7session_cassandra_private));
    if (p_st_private == NULL)
        return NULL;
    memset(p_st_private, 0x00, sizeof(L7session_cassandra_private));

    return p_st_private;
}


/*!
 * 私有信息reset
 *
 * \param p_private_session  (cassandra的私有session数据)
 *
 * \return FR_SUCC   成功\n
 *
 * \note
 *
 */
FR_STATIC int __cassandra_session_reset(void *p_private_session)
{
    L7session_cassandra_private *p_st_private = (L7session_cassandra_private*)p_private_session;

    if (p_st_private)
    {
        if (p_st_private->p_st_body_data_req)
        {
            frreport_body_free(&(p_st_private->p_st_body_data_req));
            p_st_private->p_st_body_data_req = NULL;
        }

        if (p_st_private->p_st_body_data_resp)
        {
            frreport_body_free(&(p_st_private->p_st_body_data_resp));
            p_st_private->p_st_body_data_resp = NULL;
        }

        if (p_st_private->p_req_packet)
        {
            bytebuff_free(&(p_st_private->p_req_packet));
            p_st_private->p_req_packet = NULL;
        }

        if (p_st_private->p_resp_packet)
        {
            bytebuff_free(&(p_st_private->p_resp_packet));
            p_st_private->p_resp_packet = NULL;
        }

        p_st_private->u16_req_stream = 0;
        p_st_private->u16_resp_stream = 0;
        p_st_private->i_sign = 0;
    }

    return FR_SUCC;
}

/*!
 * 判断是否是cassandra协议
 *
 * \param p_st_conn      (输入数据信息)
 * \param p_st_packet    (输入数据)
 *
 * \return APM_L7_IS_YES  属于cassandra协议\n
 *         APM_L7_IS_NO   不属于cassandra协议
 *
 * \note
 *
 */
int cassandra_detect(Conntrack *p_st_conn, Apm_packet *p_st_packet)
{
    int i_ret = 0;

    if (p_st_packet == NULL || p_st_packet->p_payload == NULL || p_st_packet->ui_payload_len < 9 || p_st_conn==NULL)
        return APM_L7_IS_NO;

    //第1位：0x04 Request frame for version4
    //       0x84 Response frame for version4
    /*
    if (((*p_st_packet->p_payload & 0xff) >> 4) != 0x00)
        goto ERROR;
    */
    switch(*p_st_packet->p_payload)
    {
        case 0x01:
        case 0x02:
        case 0x03:
        case 0x04:
            break;
        default:
            goto ERROR;
    }
    //第2位：标志位，0x01压缩标志、0x02跟踪标志、0x04自定义负载标志、0x08警告标志，其余未使用并忽略
    switch(*(p_st_packet->p_payload+1))
    {
        case 0x01:
        case 0x02:
        case 0x04:
        case 0x08:
        case 0x00:
            break;
        default:
            goto ERROR;
    }
    //第5位：opcode，request有STARTUP(0x01)、AUTH_RESPONSE(0x0F)、OPTIONS(0x05)、
    //QUERY(0x07)、PREPARE(0x09)、EXECUTE(0x0A)、BATCH(0x0D)、REGISTER(0x0B)
    switch(*(p_st_packet->p_payload+4))
    {
        case 0x01:
        case 0x05:
        case 0x07:
        case 0x09:
        case 0x0A:
        case 0x0B:
        case 0x0D:
        case 0x0F:
            break;
        default:
            goto ERROR;
    }
    //第6、7、8、9位：长度（大端）
    //arr_len[0] = *(p_st_packet->p_payload+5);
    //arr_len[1] = *(p_st_packet->p_payload+6);
    //arr_len[2] = *(p_st_packet->p_payload+7);
    //arr_len[3] = *(p_st_packet->p_payload+8);
    //p_i_tmp = (int *)arr_len;
    //if (ntohl(*p_i_tmp) != p_st_packet->ui_payload_len-9)
        //goto ERROR;

    i_ret = frct_get_direct_known(p_st_conn);
    if (i_ret == FR_DIRECT_UNKNOWN)
        frcap_set_cs_srv(p_st_conn, p_st_packet->dst_ip, p_st_packet->dst_port);
    frservice_add(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, NULL, 0);
    return APM_L7_IS_YES;

ERROR:
    if (frct_get_direct_known(p_st_conn) == FR_DIRECT_UNKNOWN)
    {
        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, NULL, 0, NULL);
        if (i_ret == FR_SUCC)
        {
            frcap_set_cs_srv(p_st_conn, p_st_packet->dst_ip, p_st_packet->dst_port);
            return APM_L7_IS_YES;
        }
        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->src_ip, p_st_packet->src_port, NULL, 0, NULL);
        if (i_ret == FR_SUCC)
        {
            frcap_set_cs_srv(p_st_conn, p_st_packet->src_ip, p_st_packet->src_port);
            return APM_L7_IS_YES;
        }
    }
    else
    {
        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->src_ip, p_st_packet->src_port, NULL, 0, NULL);
        if (i_ret == FR_SUCC)
            return APM_L7_IS_YES;

        i_ret = frservice_get(g_p_apm_context, MODULE_L7PROTO, p_st_packet->dst_ip, p_st_packet->dst_port, NULL, 0, NULL);
        if (i_ret == FR_SUCC)
            return APM_L7_IS_YES;
    }
    return APM_L7_IS_NO;
}

/*!
 * 链表节点释放规则
 *
 * \Frlist_node *p_st_node
 *
 * \return void
 *
 * \note
 *
 */
FR_STATIC void __list_node_free(Frlist_node *p_st_node)
{
    cassandra_report_list *p_st_info = NULL;

    if (p_st_node==NULL || p_st_node->data == NULL)
        return;

    p_st_info = (cassandra_report_list *)FRLIST_GET_NODE_DATA(p_st_node);

    if (p_st_info->p_lst_req)
        frreport_body_l7list_free(&p_st_info->p_lst_req);
    if (p_st_info->p_lst_comm)
        frreport_body_l7list_free(&p_st_info->p_lst_comm);

    frfree(p_st_info);
    return;
}

/*!
 * req的p_lst_comm、p_lst_req、starttime和ID存进链表
 *
 * \param p_st_private    (cassandra私有数据)
 *
 * \return FR_FAIL        失败\n
 *         FR_SUCC        成功
 *
 * \note
 *
 */
FR_STATIC int __cassandra_add_req_list(L7session_cassandra_private *p_st_private)
{
    cassandra_report_list *p_st_report = NULL;
	Frlist_node *p_st_node = NULL;
	Frlist_node *p_st_gotnode = NULL;
	L4_transaction *p_st_l4_trans = NULL;

    if (p_st_private == NULL)
        return FR_FAIL;

    //创建链表
    if (p_st_private->p_report_list_cassandra == NULL)
        p_st_private->p_report_list_cassandra = frhash2_create2(100);

    p_st_l4_trans = &(p_st_private->st_l4_trans);

    //把req的p_lst_comm、p_lst_req和ID和时间存进链表
    p_st_report = (cassandra_report_list *)frmalloc(sizeof(cassandra_report_list));
    memset(p_st_report, 0x00, sizeof(cassandra_report_list));

    p_st_report->st_request_time.tv_sec = p_st_l4_trans->st_request_time.tv_sec;
    p_st_report->st_request_time.tv_usec = p_st_l4_trans->st_request_time.tv_usec;

    //拷贝相关链表
    p_st_report->p_lst_comm = FRREPORT_BODY_L7LIST_CLONE_COMM(p_st_private->p_st_body_data_req);
    if (p_st_report->p_lst_comm == NULL)
        goto ERROR;
    p_st_report->p_lst_req = FRREPORT_BODY_L7LIST_CLONE_REQ(p_st_private->p_st_body_data_req);
    if (p_st_report->p_lst_req == NULL)
        goto ERROR;

    p_st_node = frlist_node_create(p_st_report);
    //存list之前 判断key值是否存在 如存在删除节点 再加节点
    p_st_gotnode = frhash2_getLong(p_st_private->p_report_list_cassandra, p_st_private->u16_req_stream);
    if (p_st_gotnode != NULL)
    {
        frhash2_del(p_st_private->p_report_list_cassandra, p_st_gotnode, __list_node_free);
        p_st_gotnode = NULL;
    }
    frhash2_addLong(p_st_private->p_report_list_cassandra, p_st_private->u16_req_stream, p_st_node);

    return FR_SUCC;

ERROR:
    if (p_st_report)
        frfree(p_st_report);
    if (p_st_report->p_lst_comm)
        frreport_body_l7list_free(&p_st_report->p_lst_comm);
    if (p_st_report->p_lst_req)
        frreport_body_l7list_free(&p_st_report->p_lst_req);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * req数据解析
 *
 * \param p_packet  (req数据)
 * \param p_st_private    (私有数据)
 * \param p_st_body_data
 * \param p_st_conn       (入力链接信息)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *         APM_L7_ANALYZE_TODO
 *
 * \note
 *
 */
FR_STATIC int __req_parse(Frbyte_Packet *p_packet
                        , L7session_cassandra_private *p_st_private
                        , Frreport_body_data *p_st_body_data
                        , Conntrack *p_st_conn)
{
    int i_count = 0;
    int32_t i32_value = 0;
    uint32_t u32_count = 0;
    uint32_t u32_body_len = 0;
    char arr_opcode[OPCODE_LEN];
    char arr_name[NAME_LEN];
    char *p_value = NULL;

    if (p_packet == NULL || p_st_private == NULL || p_st_body_data == NULL || p_st_conn == NULL)
        return APM_L7_ANALYZE_FAIL;

    memset(arr_opcode, 0x00, sizeof(arr_opcode));
    memset(arr_name, 0x00, sizeof(arr_name));

    if (bytebuff_remaining(p_packet) < 9)
        goto ERROR;
    //第1位：version、第2位：flag，跳过
    if (bytebuff_skip(p_packet, 2) != FR_SUCC)
        goto ERROR;
    //第3、4位：stream，保存在私有数据里
    if (bytebuff_get_short(p_packet, 2, &(p_st_private->u16_req_stream), BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;
    //第5位：opcode，
    if (bytebuff_get_bytes(p_packet, arr_opcode, 1) == FR_FAIL)
        goto ERROR;
    //第6、7、8、9位：body长度
    if (bytebuff_get_int(p_packet, 4, &u32_body_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;
    //if (u32_body_len+9 != p_packet->ui_all_len)
        //return APM_L7_ANALYZE_TODO;

    switch(*arr_opcode)
    {
        //STARTUP
        case 0x01:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "STARTUP") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 2)
                goto ERROR;
            if (bytebuff_get_int(p_packet, 2, &u32_count, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            for (i_count=0; (uint32_t)i_count<u32_count; i_count++)
            {
                if (bytebuff_remaining(p_packet) < 2)
                    return APM_L7_ANALYZE_TODO;
                if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                    goto ERROR;
                if (i32_value > 0)
                {
                    if (bytebuff_remaining(p_packet) < i32_value)
                        return APM_L7_ANALYZE_TODO;
                    p_value = (char*)frmalloc(i32_value+1);
                    if (p_value == NULL)
                        goto ERROR;
                    memset(p_value, 0x00, i32_value+1);
                    if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                        goto ERROR;
                    snprintf(arr_name, sizeof(arr_name), "request.cassandra.startup.%s", p_value);
                    frfree(p_value);
                    p_value = NULL;
                }
                else
                {
                    snprintf(arr_name, sizeof(arr_name), "request.cassandra.startup.%s", "");
                }
                if (bytebuff_remaining(p_packet) < 2)
                    return APM_L7_ANALYZE_TODO;
                if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                    goto ERROR;
                if (i32_value > 0)
                {
                    if (bytebuff_remaining(p_packet) < i32_value)
                        return APM_L7_ANALYZE_TODO;
                    p_value = (char*)frmalloc(i32_value+1);
                    if (p_value == NULL)
                        goto ERROR;
                    memset(p_value, 0x00, i32_value+1);
                    if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                        goto ERROR;
                    if (frreport_body_l7_add_str(p_st_body_data, arr_name, p_value) == FR_FAIL)
                        goto ERROR;
                    frfree(p_value);
                    p_value = NULL;
                }
                else
                {
                    if (frreport_body_l7_add_str(p_st_body_data, arr_name, "") == FR_FAIL)
                        goto ERROR;
                }
            }
            break;
        //OPTIONS
        case 0x05:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "OPTIONS") == FR_FAIL)
                goto ERROR;
            break;
        //QUERY
        case 0x07:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "QUERY") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 4, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, "request.cassandra.query", p_value) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, "request.cassandra.query", "") == FR_FAIL)
                    goto ERROR;
            }
            if ((uint32_t)bytebuff_remaining(p_packet) < u32_body_len-i32_value-4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_skip(p_packet, u32_body_len-i32_value-4) != FR_SUCC)
                goto ERROR;
            break;
        //PREPARE
        case 0x09:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "PREPARE") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 4, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, "request.cassandra.prepare", p_value) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, "request.cassandra.prepare", "") == FR_FAIL)
                    goto ERROR;
            }
            break;
        //EXECUTE
        case 0x0A:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "EXECUTE") == FR_FAIL)
                goto ERROR;
            //<id><query_parameters>
            break;
        //REGISTER
        case 0x0B:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "REGISTER") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 2)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 2, &u32_count, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            for (i_count=0; (uint32_t)i_count<u32_count; i_count++)
            {
                if (bytebuff_remaining(p_packet) < 2)
                    return APM_L7_ANALYZE_TODO;
                if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                    goto ERROR;
                if (i32_value > 0)
                {
                    if (bytebuff_remaining(p_packet) < i32_value)
                        return APM_L7_ANALYZE_TODO;
                    p_value = (char*)frmalloc(i32_value+1);
                    if (p_value == NULL)
                        goto ERROR;
                    memset(p_value, 0x00, i32_value+1);
                    if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                        goto ERROR;
                    snprintf(arr_name, sizeof(arr_name), "request.cassandra.register.%d", i_count+1);
                    if (frreport_body_l7_add_str(p_st_body_data, arr_name, p_value) == FR_FAIL)
                        goto ERROR;
                    frfree(p_value);
                    p_value = NULL;
                }
                else
                {
                    snprintf(arr_name, sizeof(arr_name), "request.cassandra.register.%d", i_count+1);
                    if (frreport_body_l7_add_str(p_st_body_data, arr_name, "") == FR_FAIL)
                        goto ERROR;
                }
            }
            break;
        //BATCH
        case 0x0D:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "BATCH") == FR_FAIL)
                goto ERROR;
            //<type><n><query_1>...<query_n><consistency><flags>[<serial_consistency>][<timestamp>]
            break;
        //AUTH_RESPONSE
        case 0x0F:
            if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_TRANS_TITLE, "AUTH_RESPONSE") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 4, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, "request.cassandra.auth_response", p_value+1) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, "request.cassandra.auth_response", "") == FR_FAIL)
                    goto ERROR;
            }
            break;
        default:
            goto ERROR;
    }
    return APM_L7_ANALYZE_SET_REQ(APM_L7_ANALYZE_SUCC);

ERROR:
    if (p_value)
        frfree(p_value);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * resp数据解析
 *
 * \param p_packet   (resp数据)
 * \param p_st_private    (私有数据)
 * \param p_st_body_data
 * \param p_st_conn       (入力链接信息)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *         APM_L7_ANALYZE_TODO
 *
 * \note
 *
 */
FR_STATIC int __resp_parse(Frbyte_Packet *p_packet
                         , L7session_cassandra_private *p_st_private
                         , Frreport_body_data *p_st_body_data
                         , Conntrack *p_st_conn)
{
    int i_count = 0;
    int i_count_tmp = 0;
    uint32_t u32_count = 0;
    uint32_t u32_list_count = 0;
    uint32_t u32_body_len = 0;
    int32_t i32_value = 0;
    char arr_opcode[OPCODE_LEN];
    char arr_name[NAME_LEN];
    char arr_name_tmp[NAME_LEN];
    char *p_value = NULL;

    if (p_packet == NULL || p_st_private == NULL || p_st_body_data == NULL || p_st_conn == NULL)
        return APM_L7_ANALYZE_FAIL;

    memset(arr_opcode, 0x00, sizeof(arr_opcode));
    memset(arr_name, 0x00, sizeof(arr_name));
    memset(arr_name_tmp, 0x00, sizeof(arr_name_tmp));

    if (bytebuff_remaining(p_packet) < 9)
        goto ERROR;
    //第1位：version、第2位：flag，跳过
    if (bytebuff_skip(p_packet, 2) != FR_SUCC)
        goto ERROR;
    //第3、4位：stream
    if (bytebuff_get_short(p_packet, 2, &(p_st_private->u16_resp_stream), BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;
    //第5位：opcode，
    if (bytebuff_get_bytes(p_packet, arr_opcode, 1) == FR_FAIL)
        goto ERROR;
    //第6、7、8、9位：body长度
    if (bytebuff_get_int(p_packet, 4, &u32_body_len, BYTE_BIG_ENDIAN) == FR_FAIL)
        goto ERROR;
    //if (u32_body_len+9 !=p_packet->ui_all_len)
        //return APM_L7_ANALYZE_TODO;

    switch(*arr_opcode)
    {
        //ERROR
        case 0x00:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, -1) == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 6)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_skip(p_packet, 4) != FR_SUCC)
                goto ERROR;
            if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_RESP_CODE_STRING, p_value) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, APM_REPORT_KEY_RESP_CODE_STRING, "") == FR_FAIL)
                    goto ERROR;
            }
            break;
        //READY
        case 0x02:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
                goto ERROR;
            if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "READY") == FR_FAIL)
                goto ERROR;
            break;
        //AUTHENTICATE
        case 0x03:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
                goto ERROR;
            if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "AUTHENTICATE") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 2)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.authenticate", p_value) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.authenticate", "") == FR_FAIL)
                    goto ERROR;
            }
            break;
        //SUPPORTED
        case 0x06:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
                goto ERROR;
            if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "SUPPORTED") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 2)
                return APM_L7_ANALYZE_TODO;
            //A [short] n, followed by n pair <k><v> where <k> is a [string] and <v> is a [string list]
            if (bytebuff_get_int(p_packet, 2, &u32_count, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            for (i_count=0; (uint32_t)i_count<u32_count; i_count++)
            {
                if (bytebuff_remaining(p_packet) < 2)
                    return APM_L7_ANALYZE_TODO;
                if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                    goto ERROR;
                if (i32_value > 0)
                {
                    if (bytebuff_remaining(p_packet) < i32_value)
                        return APM_L7_ANALYZE_TODO;
                    p_value = (char*)frmalloc(i32_value+1);
                    if (p_value == NULL)
                        goto ERROR;
                    memset(p_value, 0x00, i32_value+1);
                    if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                        goto ERROR;
                    snprintf(arr_name, sizeof(arr_name), "response.cassandra.supported.%s", p_value);
                    frfree(p_value);
                    p_value = NULL;
                }
                else
                {
                    snprintf(arr_name, sizeof(arr_name), "response.cassandra.supported.%s", "");
                }
                if (bytebuff_remaining(p_packet) < 2)
                    return APM_L7_ANALYZE_TODO;
                if (bytebuff_get_int(p_packet, 2, &u32_list_count, BYTE_BIG_ENDIAN) == FR_FAIL)
                    goto ERROR;
                for (i_count_tmp=0; (uint32_t)i_count_tmp<u32_list_count; i_count_tmp++)
                {
                    if (bytebuff_remaining(p_packet) < 2)
                        return APM_L7_ANALYZE_TODO;
                    if (bytebuff_get_int(p_packet, 2, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                        goto ERROR;
                    if (i32_value > 0)
                    {
                        if (bytebuff_remaining(p_packet) < i32_value)
                            return APM_L7_ANALYZE_TODO;
                        p_value = (char*)frmalloc(i32_value+1);
                        if (p_value == NULL)
                            goto ERROR;
                        memset(p_value, 0x00, i32_value+1);
                        if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                            goto ERROR;
                        snprintf(arr_name_tmp, sizeof(arr_name_tmp), "%s.%d", arr_name, i_count_tmp+1);
                        if (frreport_body_l7_add_str(p_st_body_data, arr_name_tmp, p_value) == FR_FAIL)
                            goto ERROR;
                        frfree(p_value);
                        p_value = NULL;
                    }
                    else
                    {
                        snprintf(arr_name_tmp, sizeof(arr_name_tmp), "%s.%d", arr_name, i_count_tmp+1);
                        if (frreport_body_l7_add_str(p_st_body_data, arr_name_tmp, "") == FR_FAIL)
                            goto ERROR;
                    }
                }
            }
            break;
        //RESULT
        case 0x08:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
                goto ERROR;
            if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "RESULT") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 4, &u32_count, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (u32_count == 1)
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.result", "void") == FR_FAIL)
                    goto ERROR;
            }
            else if (u32_count == 2)
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.result", "rows") == FR_FAIL)
                    goto ERROR;
            }
            else if (u32_count == 3)
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.result", "set_keyspace") == FR_FAIL)
                    goto ERROR;
            }
            else if (u32_count == 4)
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.result", "prepared") == FR_FAIL)
                    goto ERROR;
            }
            else if (u32_count == 5)
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.result", "schema_change") == FR_FAIL)
                    goto ERROR;

            }
            else
                goto ERROR;
            if ((uint32_t)bytebuff_remaining(p_packet) < u32_body_len-4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_skip(p_packet, u32_body_len-4) != FR_SUCC)
                goto ERROR;
            break;
        //EVENT
        case 0x0C:
            p_st_private->i_sign = 1;
            return APM_L7_ANALYZE_TODO;
            //if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "EVENT") == FR_FAIL)
                //goto ERROR;
            break;
        //AUTH_CHALLENGE
        case 0x0E:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
                goto ERROR;
            if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "AUTH_CHALLENGE") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 4, (uint32_t *)&i32_value, BYTE_BIG_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.auth_challenge", p_value) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.auth_challenge", "") == FR_FAIL)
                    goto ERROR;
            }
            break;
        //AUTH_SUCCESS
        case 0x10:
            if (frreport_body_l7_add_int(p_st_body_data, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
                goto ERROR;
            if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.opcode", "AUTH_SUCCESS") == FR_FAIL)
                goto ERROR;
            if (bytebuff_remaining(p_packet) < 4)
                return APM_L7_ANALYZE_TODO;
            if (bytebuff_get_int(p_packet, 4, (uint32_t *)&i32_value, BYTE_LITTLE_ENDIAN) == FR_FAIL)
                goto ERROR;
            if (i32_value > 0)
            {
                if (bytebuff_remaining(p_packet) < i32_value)
                    return APM_L7_ANALYZE_TODO;
                p_value = (char*)frmalloc(i32_value+1);
                if (p_value == NULL)
                    goto ERROR;
                memset(p_value, 0x00, i32_value+1);
                if (bytebuff_get_bytes(p_packet, p_value, i32_value) == FR_FAIL)
                    goto ERROR;
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.auth_success", p_value) == FR_FAIL)
                    goto ERROR;
                frfree(p_value);
            }
            else
            {
                if (frreport_body_l7_add_str(p_st_body_data, "response.cassandra.auth_success", "") == FR_FAIL)
                    goto ERROR;
            }
            break;
        default:
            goto ERROR;
    }
    return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);

ERROR:
    if (p_value)
        frfree(p_value);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * 检查是否超时
 *
 * \param p_node
 * \param p_data
 *
 * \return int
 *
 * \note
 *
 */
FR_STATIC int __check_timeout(Frlist_node* p_node, const void *p_data)
{
    cassandra_report_list *p_st_info = NULL;
    Frhash_header *p_header = (Frhash_header *)p_data;

    p_st_info = (cassandra_report_list *)p_node->data;

    if (p_st_info->st_request_time.tv_sec + SIP_TIMEOUT <= g_st_time.tv_sec)
    {
        frhash2_del(p_header, p_node, __list_node_free);
        return 1;
    }

    return 1;
}

/*!
 * 清除hash中超时的node
 *
 * \param p_hash_stmt
 * \param st_timestamp
 *
 * \return int
 *
 * \note
 *
 */
FR_STATIC void __clear_timeout_node(Frhash_header *p_header, struct timeval st_timestamp)
{
    g_st_time = st_timestamp;
    frhash2_run(p_header, p_header, __check_timeout);

    return;
}

/*!
 * cassandra的packet解析
 *
 * \param p_st_conn          (入力链接信息)
 * \param p_st_packet       (入力packet信息)
 * \param p_st_hash_report   (保存Frreport_data)
 * \param p_st_session       (私有数据)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *         APM_L7_ANALYZE_TODO
 *
 * \note
 *
 */
FR_STATIC int __req_data_analysis(Conntrack *p_st_conn
                                , Apm_packet *p_st_packet
                                , Frhash_header *p_st_hash_report
                                , L7_session *p_st_session)
{
    int i_ret = 0;
    unsigned long ul_tmp = 0;
    L4_transaction *p_st_l4_trans = NULL;
    Frreport_data* p_st_report_data = NULL;
    L7session_cassandra_private *p_st_private = NULL;

    if (p_st_conn == NULL || p_st_packet == NULL || p_st_hash_report == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    p_st_private = (L7session_cassandra_private*)p_st_session->p_session_private;
    p_st_l4_trans = &(p_st_private->st_l4_trans);

    //创建数据链表
    if (p_st_private->p_st_body_data_req == NULL)
    {
        p_st_private->p_st_body_data_req = frreport_body_new_l7(p_st_conn);
        if (p_st_private->p_st_body_data_req == NULL)
            goto ERROR;

        if (frreport_body_l7_add_connInfo(p_st_private->p_st_body_data_req, p_st_conn) == FR_FAIL)
            goto ERROR;

        if (p_st_l4_trans->st_request_time_first_byte.tv_sec==0 && p_st_l4_trans->st_request_time_first_byte.tv_usec==0)
        {
            //st_request_time_first_byte
            p_st_l4_trans->st_request_time_first_byte.tv_sec = p_st_packet->st_timestamp.tv_sec;
            p_st_l4_trans->st_request_time_first_byte.tv_usec = p_st_packet->st_timestamp.tv_usec;
        }

        //创建bytebuff
        if (p_st_private->p_req_packet == NULL)
        {
            p_st_private->p_req_packet = bytebuff_creat();
            if (p_st_private->p_req_packet == NULL)
                goto ERROR;
        }
    }
    //请求的字节数
    p_st_l4_trans->l_request_bytes += p_st_packet->ui_payload_len;

    if (p_st_private->p_req_packet == NULL)
        goto ERROR;
    //把这个包的数据先全加进bytebuff里
    if (bytebuff_add(p_st_private->p_req_packet, p_st_packet->p_payload, p_st_packet->ui_payload_len) != FR_SUCC)
        goto ERROR;

    while(bytebuff_remaining(p_st_private->p_req_packet) > 5)
    {
        //创建数据链表
        if (p_st_private->p_st_body_data_req == NULL)
        {
            p_st_private->p_st_body_data_req = frreport_body_new_l7(p_st_conn);
            if (p_st_private->p_st_body_data_req == NULL)
                goto ERROR;
            if (frreport_body_l7_add_connInfo(p_st_private->p_st_body_data_req, p_st_conn) == FR_FAIL)
                goto ERROR;
        }
        
        //req解析
        i_ret = __req_parse(p_st_private->p_req_packet, p_st_private, p_st_private->p_st_body_data_req, p_st_conn);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;
        if (i_ret == APM_L7_ANALYZE_TODO)
        {
            frreport_body_free(&p_st_private->p_st_body_data_req);
            bytebuff_rewind(p_st_private->p_req_packet);
            return i_ret;
        }

        //数据上报
        if (frreport_body_l7_add_ulong(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_ID, frct_seq_increase(p_st_conn)) == FR_FAIL)
            goto ERROR;

        //start time
        if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_START_TIMESTAMP, &(p_st_l4_trans->st_request_time_first_byte)) == FR_FAIL)
            goto ERROR;

        //st_request_time
        p_st_l4_trans->st_request_time.tv_sec = p_st_packet->st_timestamp.tv_sec;
        p_st_l4_trans->st_request_time.tv_usec = p_st_packet->st_timestamp.tv_usec;
        ul_tmp = frtime_diff(&(p_st_l4_trans->st_request_time_first_byte), &(p_st_l4_trans->st_request_time));
        //send_time_us
        if (frreport_body_l7_add_long(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_SEND_TIME_US, ul_tmp) == FR_FAIL)
            goto ERROR;

        //request bytes
        if (frreport_body_l7_add_long(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_DATA_BYTES, p_st_l4_trans->l_request_bytes) == FR_FAIL)
            goto ERROR;

        //end time
        if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_req, APM_REPORT_KEY_REQ_END_TIMESTAMP, &p_st_l4_trans->st_request_time) == FR_FAIL)
            goto ERROR;

        p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_REQ);
        if (p_st_report_data == NULL)
            goto ERROR;

        //p_lst_req p_lst_comm添加到链表
        i_ret = __cassandra_add_req_list(p_st_private);
        if (i_ret == FR_FAIL)
            goto ERROR;

        __clear_timeout_node(p_st_private->p_report_list_cassandra, p_st_packet->st_timestamp);

        if (frreport_data_add_body(p_st_report_data, p_st_private->p_st_body_data_req) == FR_FAIL)
            goto ERROR;
        p_st_private->p_st_body_data_req = NULL;

        if (frreport_data_hash(p_st_hash_report, p_st_report_data) == FR_FAIL)
            goto ERROR;
        p_st_report_data = NULL;
    }

    __cassandra_session_reset(p_st_private);

    return APM_L7_ANALYZE_SET_REQ(APM_L7_ANALYZE_SUCC);

ERROR:
    if (p_st_report_data)
        frreport_data_free(&p_st_report_data);
    __cassandra_session_reset(p_st_private);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * cassandra的packet解析
 *
 * \param p_st_conn          (入力链接信息)
 * \param p_st_packet        (入力packet信息)
 * \param p_st_hash_report   (保存Frreport_data)
 * \param p_st_session       (私有数据)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *         APM_L7_ANALYZE_TODO
 *
 * \note
 *
 */
FR_STATIC int __resp_data_analysis(Conntrack *p_st_conn
                                 , Apm_packet *p_st_packet
                                 , Frhash_header *p_st_hash_report
                                 , L7_session *p_st_session)
{
    int i_ret = 0;
    unsigned long ul_tmp = 0;
    L7session_cassandra_private *p_st_private = NULL;
    L4_transaction *p_st_l4_trans = NULL;
    Frreport_data* p_st_report_data = NULL;
    Frlist_node *p_st_gotnode = NULL;
    struct timeval st_request_time;

    if (p_st_conn == NULL || p_st_packet == NULL || p_st_hash_report == NULL || p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    p_st_private = (L7session_cassandra_private*)p_st_session->p_session_private;
    p_st_l4_trans = &(p_st_private->st_l4_trans);

    if (p_st_private->p_st_body_data_resp == NULL)
    {
        //创建数据链表
        p_st_private->p_st_body_data_resp = frreport_body_new_l7(p_st_conn);
        if (p_st_private->p_st_body_data_resp == NULL)
            goto ERROR;

        if (p_st_l4_trans->st_response_time_first_byte.tv_sec == 0 && p_st_l4_trans->st_response_time_first_byte.tv_usec == 0)
        {
            p_st_l4_trans->st_response_time_first_byte.tv_sec = p_st_packet->st_timestamp.tv_sec;
            p_st_l4_trans->st_response_time_first_byte.tv_usec = p_st_packet->st_timestamp.tv_usec;
        }

        //创建bytebuff
        if (p_st_private->p_resp_packet == NULL)
        {
            p_st_private->p_resp_packet = bytebuff_creat();
            if (p_st_private->p_resp_packet == NULL)
                goto ERROR;
        }
    }

    //请求的字节数
    p_st_l4_trans->l_response_bytes += p_st_packet->ui_payload_len;

    if (p_st_private->p_resp_packet == NULL)
        goto ERROR;
    //把这个包的数据先全加进bytebuff里
    if (bytebuff_add(p_st_private->p_resp_packet, p_st_packet->p_payload, p_st_packet->ui_payload_len) != FR_SUCC)
        goto ERROR;

    while(bytebuff_remaining(p_st_private->p_resp_packet) > 5)
    {
        if (p_st_private->p_st_body_data_resp == NULL)
        {
            //创建数据链表
            p_st_private->p_st_body_data_resp = frreport_body_new_l7(p_st_conn);
            if (p_st_private->p_st_body_data_resp == NULL)
                goto ERROR;
        }

        //resp解析
        i_ret = __resp_parse(p_st_private->p_resp_packet, p_st_private, p_st_private->p_st_body_data_resp, p_st_conn);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;
        if (i_ret == APM_L7_ANALYZE_TODO && p_st_private->i_sign == 1)
            return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);
        if (i_ret == APM_L7_ANALYZE_TODO)
        {
            frreport_body_free(&p_st_private->p_st_body_data_resp);
            p_st_private->p_st_body_data_resp = NULL;
            bytebuff_rewind(p_st_private->p_resp_packet);
            return i_ret;
        }

        //取链表对应的request p_lst_req p_lst_comm
        p_st_gotnode = frhash2_getLong(p_st_private->p_report_list_cassandra, p_st_private->u16_resp_stream);
        if (p_st_gotnode == NULL)
            goto ERROR;
        //数据上报
        FRREPORT_BODY_L7LIST_SET_REQ(p_st_private->p_st_body_data_resp, &(((cassandra_report_list *)(p_st_gotnode->data))->p_lst_req));
        FRREPORT_BODY_L7LIST_SET_COMM(p_st_private->p_st_body_data_resp, &(((cassandra_report_list *)(p_st_gotnode->data))->p_lst_comm));
        ((cassandra_report_list *)(p_st_gotnode->data))->p_lst_req = NULL;
        ((cassandra_report_list *)(p_st_gotnode->data))->p_lst_comm = NULL;

        //start time
        if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_START_TIMESTAMP, &(p_st_l4_trans->st_response_time_first_byte)) == FR_FAIL)
            goto ERROR;

        st_request_time.tv_sec = ((cassandra_report_list *)(p_st_gotnode->data))->st_request_time.tv_sec;
        st_request_time.tv_usec = ((cassandra_report_list *)(p_st_gotnode->data))->st_request_time.tv_usec;

        frhash2_del(p_st_private->p_report_list_cassandra, p_st_gotnode, __list_node_free);

        //server time
        if (st_request_time.tv_sec != 0 || st_request_time.tv_usec != 0)
        {
            ul_tmp = frtime_diff(&(st_request_time), &(p_st_l4_trans->st_response_time_first_byte));
            if (frreport_body_l7_add_ulong(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_SERVER_TIME_US, ul_tmp) == FR_FAIL)
                goto ERROR;
        }
        else
        {
            if (frreport_body_l7_add_ulong(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_SERVER_TIME_US, 0) == FR_FAIL)
                goto ERROR;
        }

        //st_response_time
        p_st_l4_trans->st_response_time.tv_sec = p_st_packet->st_timestamp.tv_sec;
        p_st_l4_trans->st_response_time.tv_usec = p_st_packet->st_timestamp.tv_usec;

        if (frreport_body_l7_add_long(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_DATA_BYTES, p_st_l4_trans->l_response_bytes) == FR_FAIL)
            goto ERROR;

        //down load time
        ul_tmp = frtime_diff(&(p_st_l4_trans->st_response_time_first_byte), &(p_st_l4_trans->st_response_time));
        if (frreport_body_l7_add_long(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_DOWNLOAD_TIME_US, ul_tmp) == FR_FAIL)
            goto ERROR;

        //end time
        if (frreport_body_l7_add_timeval(p_st_private->p_st_body_data_resp, APM_REPORT_KEY_RESP_END_TIMESTAMP, &(p_st_l4_trans->st_response_time)) == FR_FAIL)
            goto ERROR;

        p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_RESP);
        if (p_st_report_data == NULL)
            goto ERROR;

        if (frreport_data_add_body(p_st_report_data, p_st_private->p_st_body_data_resp) == FR_FAIL)
            goto ERROR;
        p_st_private->p_st_body_data_resp = NULL;

        if (frreport_data_hash(p_st_hash_report, p_st_report_data) == FR_FAIL)
            goto ERROR;
        p_st_report_data = NULL;
    }

    __cassandra_session_reset(p_st_private);

    return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);

ERROR:
    if (p_st_report_data)
        frreport_data_free(&p_st_report_data);
    __cassandra_session_reset(p_st_private);
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * cassandra的packet解析
 *
 * \param p_st_conn          (入力链接信息)
 * \param p_st_packet        (入力packet信息)
 * \param p_st_hash_report   (保存Frreport_data)
 *
 * \return APM_L7_ANALYZE_SUCC   成功\n
 *         APM_L7_ANALYZE_FAIL   失败
 *
 * \note
 *
 */
int cassandra_analysis(Conntrack *p_st_conn
                           , Apm_packet *p_st_packet
                           , Frhash_header *p_st_hash_report)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    L7_session *p_st_session = NULL;

    if (p_st_conn == NULL || p_st_packet == NULL
        || p_st_hash_report == NULL)
        return APM_L7_ANALYZE_FAIL;

    //到Conntrack中取得session信息
    p_st_session = frct_get_session(p_st_conn);
    if (p_st_session == NULL)
        return APM_L7_ANALYZE_FAIL;

    //fin、syn包则销毁私有数据
    if (p_st_packet->em_analyz_cmd == FRANALYZ_CMD_FIN || p_st_packet->em_analyz_cmd == FRANALYZ_CMD_SYN)
    {
        cassandra_session_destroy((void *)(p_st_session->p_session_private));
        p_st_session->p_session_private = NULL;
        return APM_L7_ANALYZE_SUCC;
    }

    //p_session_private分配内存
    if (p_st_session->p_session_private == NULL)
    {
        p_st_session->p_session_private = __cassandra_private_create();
        //内存分配失败
        if (p_st_session->p_session_private == NULL)
            return APM_L7_ANALYZE_FAIL;
    }

    //request
    if (p_st_packet->i_direction == FR_DIRECT_CLIENT2SRV)
    {
        i_ret = __req_data_analysis(p_st_conn,p_st_packet,p_st_hash_report,p_st_session);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;

        return i_ret;
    }
    //response
    else
    {
        i_ret = __resp_data_analysis(p_st_conn,p_st_packet,p_st_hash_report, p_st_session);
        if (i_ret == APM_L7_ANALYZE_FAIL)
            goto ERROR;

        return i_ret;
    }

ERROR:
    if (p_st_session->p_session_private)
        cassandra_session_destroy(p_st_session->p_session_private);
    p_st_session->p_session_private = NULL;
    return APM_L7_ANALYZE_FAIL;
}

/*!
 * 私有信息销毁
 *
 * \param p_private_session  (cassandra的私有session数据)
 *
 * \return FR_SUCC   成功\n
 *
 * \note
 *
 */
int cassandra_session_destroy(void *p_private_session)
{
    L7session_cassandra_private *p_st_private = (L7session_cassandra_private*)p_private_session;

    if (p_st_private)
    {
        if (p_st_private->p_st_body_data_req)
        {
            frreport_body_free(&(p_st_private->p_st_body_data_req));
            p_st_private->p_st_body_data_req = NULL;
        }

        if (p_st_private->p_st_body_data_resp)
        {
            frreport_body_free(&(p_st_private->p_st_body_data_resp));
            p_st_private->p_st_body_data_resp = NULL;
        }

        if (p_st_private->p_req_packet)
        {
            bytebuff_free(&(p_st_private->p_req_packet));
            p_st_private->p_req_packet = NULL;
        }

        if (p_st_private->p_resp_packet)
        {
            bytebuff_free(&(p_st_private->p_resp_packet));
            p_st_private->p_resp_packet = NULL;
        }

        if (p_st_private->p_report_list_cassandra)
        {
            frhash2_free2(p_st_private->p_report_list_cassandra, __list_node_free);
            p_st_private->p_report_list_cassandra = NULL;
        }

        p_st_private->u16_req_stream = 0;
        p_st_private->u16_resp_stream = 0;
        p_st_private->i_sign = 0;

        frfree(p_st_private);
        p_st_private = NULL;
    }

    return FR_SUCC;
}
